//
// file : inet_win.c
// date : 07/05/07
//
#define _WIN32_WINNT (0x0500)

#include "inet_ipc.h"
#include "inet_tmr.h"
#include <windows.h>
#include <mswsock.h>

#pragma comment(lib, "ws2_32")

/////////////////////////////////////////////////////////////////////////////

// structure used to pass parameters to user apc

typedef struct 
{
	HANDLE ready;   // event handle
	long   long0;   // return value
    long   long1;   // argument 1
	long   long2;   // argument 2

} apc_s4s4s4_t;

/////////////////////////////////////////////////////////////////////////////

void __stdcall ipc_print(char *str, ...)
{
	char        tmpstr[1024];
    SYSTEMTIME  st;
    DWORD       len;
	va_list     marker;

    GetLocalTime(&st);

    len = sprintf(tmpstr, "%.2d:%.2d:%.2d.%.3d ",
                  st.wHour, st.wMinute, st.wSecond, st.wMilliseconds);

    va_start(marker, str);

    _vsnprintf(tmpstr+len, sizeof(tmpstr)-len, str, marker);

    va_end(marker);

    /*
    ** OutputDebugStringA() may have a 1024 output chars limit
    */
    OutputDebugStringA(tmpstr);
}

//
//  handle socket events for a domain
//
int  win_soevent( ipccb_t *ipc, WSANETWORKEVENTS *w )
{
	int  e = 0;

    // make sure the order of these if-statements is correct

    if( w->lNetworkEvents & FD_CONNECT )
	{
        if( w->iErrorCode[FD_CONNECT_BIT] )
		{
			// connect errors:
            // (1) rejected - WSAECONNREFUSED
            // (2) no route - WSAENETUNREACH
            // (3) timeout  - WSAENOBUFS
            // (4) no buffer- WSAETIMEDOUT
            // action: close socket and retry
            e += -1;
		}
        else // post-connect actions
        {
            ipc->states |= (IPC_STAT_CONNECT);  // set

            if( ipc->tready )
            {
                e = ipc->tready( ipc );
            }
            // windows will send a FD_WRITE so we don't call writer here
        }
	}

    if( w->lNetworkEvents & FD_ACCEPT )
	{
        if( w->iErrorCode[FD_ACCEPT_BIT] )
		{
			// accept errors:
            // net down  - WSAENETDOWN
            // action: close socket and retry
            e += (-2);
		}
        else
        {
            // clone ipc context and put new socket
            e = ipc_accept_( ipc );  // e > 0 if succeeded

            // call	connection handler
            if( e > 0 )
            {
                // if tready defined, don't call writer
                if(	ipc->tready	)
                {
                    e = ipc->tready( ipc ); 
                }
                // windows will send a FD_WRITE so we don't call writer here
            }
        }
	}

    if( w->lNetworkEvents & FD_READ )
	{
        if( w->iErrorCode[FD_READ_BIT] )
		{
			// read errors:
            // net down  - WSAENETDOWN
            // action: close socket and retry
            e += (-4);
		}
        else
        {
            e = ipc_recving( ipc );
        }
	}

    if( w->lNetworkEvents & FD_WRITE )
	{
        if( w->iErrorCode[FD_WRITE_BIT] )
		{
			// write error:
            // net down  - WSAENETDOWN
            // action: close socket and retry
            e += (-8);
		}
        else 
        {
            // send it
            e = ipc_sending( ipc );
        }
	}

	// FD_CLOSE may comes together with FD_READ 
    // we don't receive this if we close the socket actively
    if( w->lNetworkEvents & FD_CLOSE )
	{
        if( w->iErrorCode[FD_CLOSE_BIT] )
		{
			// close errors:
            // net down   - WSAENETDOWN
            // peer reset - WSAECONNRESET
            // aborted    - WSAECONNABORTED
            e += (-16);
		}
        else
        {
            e += (-32);  // peer disconnected
        }
	}

    // FD_OOB
    // FD_ADDRESS_LIST_CHANGE
    // FD_ROUTING_INTERFACE_CHANGE

    return e;
}

//
// per session socket event
//
void net_signals( ipc_domain_t *domain )
{
    int               k;
	int               e;
    WSANETWORKEVENTS  w;

	for( k = 0; k < domain->ncbs; k++ )
	{
		ipccb_t *ipc = domain->ipcs[k];

		if(  ipc == 0 ) continue;

		if( ipc->states == IPC_STAT_INVALID ) continue;

		if( ipc->socket == INVALID_SOCKET ) continue;

		e = WSAEnumNetworkEvents( ipc->socket, 0, &w );

		if( (e == 0) && (w.lNetworkEvents != 0) )
		{
            e = win_soevent( ipc, &w );

			domain->last = k;
		}
        
        // check socket health
        if(  e == INVALID_SOCKET || e < 0)
        {
            ipc_closing( ipc );
        }
	}
}


//
// per domain master timer
//
long  __stdcall tmr_trigger( void/*uthead*/ *thead, long ms_due )
{
    LARGE_INTEGER  qw_due;
    long           period;

    ipc_domain_t  *domain;
    
    qw_due.LowPart = qw_due.HighPart = 0;
    
    period = ((uthead_t*)thead)->beat_;

    //ipc_print(" tmr_trigger (%d %d)\n", ms_due, period);

    if( ms_due > 0)  // one-time timer
    {
        if( period > 0 ) return 0; // periodic timer already on

        qw_due.LowPart = ms_due;
    }
    else if( period > 0 )// start a periodic timer
    {
        qw_due.LowPart = period;  // in milliseconds
    }
    else
    {
        return -1; // invalid parameters
    }

    assert( qw_due.LowPart > 0 );

    *(__int64 *)&qw_due *= (-10000);  // in 100-nanosecond

    domain = ((uthead_t*)thead)->owner;

    //
    // set waitable
    // NOTE: in windows we may lose 1-2 milliseconds everytime we reset timer
    //
    if( ! SetWaitableTimer(domain->wait[e_tmr_calls], &qw_due, period, 0, 0, 0) )
    {
        ipc_print("tmr_trigger() SetWaitableTimer(err=%d)\n", GetLastError());
    }

    return period;
}

//
// wait and process network/user/timer events
//
long  __stdcall ipc_waitall(ipc_domain_t *domain)
{
    ulong         e;

	if( domain == 0 ) return -1900;

	while( 0 == domain->quit )
	{
		// waiting on (1) socket events, (2) out-thread calls and (3) timer events
		e = WaitForMultipleObjectsEx(e_max_calls, domain->wait, FALSE, 1250, TRUE );

		if( e == WAIT_OBJECT_0 + e_net_calls )
		{
			net_signals( domain );  // socket events
		}

		if( e == WAIT_OBJECT_0 + e_usr_calls )
		{
			usr_signals( domain );  //  out-thread calls
		}

		if( e == WAIT_OBJECT_0 + e_tmr_calls ) 
		{
			tmr_signals( domain->qtmr );  // timer events
		}
	}

    ipc_print(" ipc_waitall(didx=%d) terminating..\n", domain->didx);

    // clear all timer  - not really needed
    tmr_stopall( domain->qtmr );

	// do cleanup
	{
		int k;

		for( k = 0; k < domain->ncbs; k++ )
		{
			ipccb_t *ipc = domain->ipcs[k];

			if( ipc == 0 ) continue;
			if( ipc->states == IPC_STAT_INVALID ) continue;
			if( ipc->socket != INVALID_SOCKET )
			{
				// de-select
                WSAEventSelect(ipc->socket, 0, 0);

                // don't call ipc_onclose() because we are final
				closesocket(ipc->socket);

				ipc->socket  = INVALID_SOCKET;
			}

            ipc->states = IPC_STAT_INVALID;
		}

        for( k = 0; k < e_max_calls; k++ ) CloseHandle( domain->wait[k] );
	}

	return 0;
}


//
// set socket events handlers
// for windows, call WSAEventSelect()
// for linux, set realtime SIGIO handler
// for bsd/unix, set SIGIO handler
//
long  __stdcall ipc_hookup_( ipc_domain_t *domain, ipccb_t *ipc )
{
	int           e;
    ulong         flags;

	flags = 0;

	if( ipc->sotype == SOCK_DGRAM )           // udp
	{
        unsigned long disable=FALSE;

        // ignore UDP PORT_UNREACHABLE 
        // defined in <mswsock.h>
        ioctlsocket(ipc->socket, SIO_UDP_CONNRESET, &disable);

		flags |= (FD_CLOSE|FD_READ|FD_WRITE);
	}
	else if( ipc->sotype == SOCK_STREAM )
	{
		if( ipc->sflags & SOC_FLAG_PASSIVE )  // tcp server
		{
            flags |= (FD_CLOSE|FD_ACCEPT|FD_READ|FD_WRITE); 
		}
	    else                                  // tcp client
	    {
		    flags |= (FD_CLOSE|FD_READ|FD_WRITE|FD_CONNECT);
	    }
	}

    e = WSAEventSelect( ipc->socket, domain->wait[0], flags);

	return 0;
}


//
// windows specific initialization for ipc domain
//
long  __stdcall ipc_osinit_(ipc_domain_t *domain)
{
    // a handle for socket event
    if( domain->wait[e_net_calls] == 0 )
    {
        domain->wait[e_net_calls] = CreateEvent(0, 0, 0, 0);  // need auto-reset
    }

    // a handle for user apc
    if( domain->wait[e_usr_calls] == 0 )
    {
        domain->wait[e_usr_calls] = CreateEvent(0, 0, 0, 0);  // need auto-reset
    }

    // a waitable timer
    if( domain->wait[e_tmr_calls] == 0 )
    {
        domain->wait[e_tmr_calls] = CreateWaitableTimer(0, 0, 0);
    }

	return 0;
}

//
// wait for apc to complete TODO : error handling
//
long  __stdcall ipc_waitapc(ipc_domain_t *domain, ipccb_t *ipc)
{
    // apcfun must be set
	if( ipc->apcfun == 0 ) return 0;

	if( ipc->apcsem == (void*)~0 )
	{
	    ipc->apcsem = CreateEvent(0, 0, 0, 0);
	}
	SetEvent( domain->wait[1] );

    SwitchToThread(); // kick our domain thread, in case of single cpu

	if( ipc->apcsem != (void*)~0 )
	{
		WaitForSingleObject( ipc->apcsem, INFINITE);

		CloseHandle(ipc->apcsem);

		ipc->apcsem = (void*)~0;

	    return ipc->apcval;
	}

    return 0; // don't wait
}

//
// apc execution thread routine
//
long  __stdcall ipc_thread_(void *par)
{
    apc_s4s4s4_t  *args;
	ipc_domain_t  *domain;
    WSADATA       wsa;

	domain = 0;

	if( (args = (apc_s4s4s4_t*)par) == 0 ) return 0;

    WSAStartup(MAKEWORD(2,0), &wsa);

    args->long0 = ipc_initial( args->long1, args->long2 );  // create a domain

	if( args->long0 > 0 ) // domain id must be positive
	{
		domain = get_domain_( args->long0 );
	}

	if( args->ready ) SetEvent( args->ready ); // notify caller

	if( domain )
	{
        // block until finish
	    ipc_waitall( domain );
	}

    WSACleanup();

    if( domain->sync ) SetEvent( domain->sync );

	return 0;
}

//
// start an ipc domain thread, return domain index
// ms_beat = 0 : no heart beat
// ms_beat < 0 : default heart beat (TMR_DEFAULT_BEAT in inet_tmr.h)
// ms_beat > 0 : heart beat rate = ms_beat
//
long  __stdcall ipc_startup(long max_sess, long ms_beat)
{
	HANDLE        h;
    DWORD         d;
	apc_s4s4s4_t  args;

    if( max_sess <= 1 ) max_sess = 1;

    memset( &args, 0, sizeof(args));

	args.ready = CreateEvent(0, 0, 0, 0);
	args.long1 = max_sess;
    args.long2 = ms_beat;

	h = CreateThread(0, 0, ipc_thread_, &args, 0, &d);

	CloseHandle( h );

	WaitForSingleObject( args.ready, INFINITE );

	CloseHandle( args.ready );

	return args.long0;  // domain index
}


//
//
//
long  __stdcall ipc_osexit_( ipc_domain_t *domain )
{

    domain->sync = CreateEvent(0, 0, 0, 0);

    domain->quit = ~0;  // exit indication

    SetEvent( domain->wait[e_usr_calls] ); // wake up 

    WaitForSingleObject( domain->sync, 5000 );

    CloseHandle( domain->sync );

    domain->sync = 0;

    return 0;
}
